package com.order.service.impl;

import com.alibaba.fastjson.JSON;
import com.order.service.OrderService;
import com.order.exception.CustomException;
import com.order.dao.OrderDao;
import com.order.entity.Order;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.cache.annotation.CacheEvict;
import org.springframework.cache.annotation.Cacheable;
import org.springframework.context.annotation.Primary;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Map;

/**
 * @author
 */
@Service
@Primary //优先注入这个bean
public class OrderServiceImpl implements OrderService {
	@Autowired
	private OrderDao orderDao;
	private Logger logger = LoggerFactory.getLogger(getClass());
	@Autowired
    private RedisTemplate redisTemplate;
	@Resource
	private KafkaTemplate<String,Object> kafkaTemplate;

	/*
    使用通用mymapper进行增删改查
    */


    @Override
	public Map<String, Object> insert(Order entity, String sessionId)throws CustomException {
		Map<String, Object> result = new HashMap<>();
		try{
			orderDao.insert(entity);
		}catch (DuplicateKeyException e){
			logger.error("no重复");
			throw new CustomException("订单号重复",CustomException.ExceptionName.RepeatFailed);
		}
		redisTemplate.opsForValue().set("order"+entity.getNo(),JSON.toJSONStringWithDateFormat(entity,"yyyy-MM-dd HH:mm:ss"));
		result.put("code","200");
		result.put("message","insert success");
		result.put("data",entity);
		logger.info("新增的订单为:{}",JSON.toJSONStringWithDateFormat(entity,"yyyy-MM-dd HH:mm:ss"));
		return result;
	}


	@Override
	@CacheEvict(cacheNames = "no",key = "'order'+#no")
	public Map<String, Object> delete(Integer no, String sessionId)throws CustomException {
		Map<String, Object> result = new HashMap<>();
		if (redisTemplate.opsForValue().get("order"+no)==null) {
			logger.error("订单号不存在,no:{}", no);
            throw new CustomException("订单号不存在",CustomException.ExceptionName.ExistFailed);
		} else {
		orderDao.deleteByNo(no);
		redisTemplate.delete("order" + no);
		logger.info("删除的订单为:{}", no);
		result.put("code", "200");
		result.put("message","delete sucess");
	}
		return result;
	}

	@Override
	public Map<String, Object> update(Order entity, String sessionId)throws CustomException {
		Map<String, Object> result = new HashMap<>();
		Map<String ,Object> no= new HashMap<>();
		no.put("no",entity.getNo());
		if(redisTemplate.opsForValue().get("order"+entity.getNo())==null){
			logger.error("修改的订单号不存在,no:{}",entity.getNo());
			throw new CustomException("订单号不存在",CustomException.ExceptionName.ExistFailed);
		}else{
		orderDao.updateByPrimaryKey(entity);
		redisTemplate.opsForValue().setIfPresent("order"+entity.getNo(),JSON.toJSONStringWithDateFormat(entity,"yyyy-MM-dd HH:mm:ss"));
		result.put("code","200");
		result.put("message","update sucess");
		result.put("data",entity);
		return result;
		}
	}


	@Override
    @Cacheable(cacheNames = "no",key = "'order'+#no",unless = "#result == null")
	public Map<String,Object> selectByNo(Integer no, String sessionId)throws CustomException {
		logger.info("查询no为:{}",no);
		return orderDao.selectByNo(no);
	}

    /*
    通过kafka上传订单,消费订单
     */
	@Override
	public Map<String,Object>send(Order entity, String sessionId)throws CustomException {
		Map<String, Object> result = new HashMap<>();
		Map<String, Object> no = new HashMap<>();
		no.put("no", entity.getNo());
		if (redisTemplate.opsForValue().get("order"+entity.getNo())==null) {
			kafkaTemplate.send("test", JSON.toJSONStringWithDateFormat(entity,"yyyy-MM-dd HH:mm:ss"));
			result.put("code","200");
			result.put("message","insert success");
			result.put("data",entity);
			return result;
		} else {
			logger.error("no重复");
			result.put("data", "no重复,不向kafka发送消息");
			throw new CustomException("订单号重复", CustomException.ExceptionName.RepeatFailed);
		}
	}

    @Override
	@KafkaListener(topics = "test")
	public void consumer(String message) throws CustomException{
			orderDao.insert (JSON.parseObject(message, Order.class));
			redisTemplate.opsForValue().set("order"+JSON.parseObject(message, Order.class).getNo(),message);
		    logger.info("新增的订单为:{}",message);
	}

}
